package io.github.quickmsg.common.channel;

import com.fasterxml.jackson.annotation.JsonIgnore;
import io.github.quickmsg.common.context.ContextHolder;
import io.github.quickmsg.common.enums.DisconnectStatus;
import io.github.quickmsg.common.handler.MyInboundIdleStateHandler;
import io.github.quickmsg.common.integrate.SubscribeTopic;
import io.github.quickmsg.common.integrate.cache.ConnectCache;
import io.github.quickmsg.common.message.mqtt.PublishMessage;
import io.github.quickmsg.common.message.mqtt.RetryMessage;
import io.github.quickmsg.common.retry.RetryManager;
import io.github.quickmsg.common.utils.MqttMessageUtils;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import lombok.*;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;
import reactor.netty.Connection;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

/**
 * @author luxurong
 */
@Getter
@Setter
@Slf4j
@EqualsAndHashCode
public class MqttChannel {

    @JsonIgnore
    private static final int MAX_MESSAGE_ID = 65535;

    /**
     * <消息id, 消息>
     */
    private Map<Integer, PublishMessage> qosCache = new HashMap<>();

    /**
     * channelId
     */
    private Long id;
    /**
     * clientId
     */
    private String clientId;
    /**
     * channel 初始化的时间
     */
    private long activeTime;
    /**
     * 授权成功时间
     */
    private String authTime;
    /**
     * 连接缓存
     */
    private ConnectCache connectCache;
    /**
     * 客户端地址
     */
    private String address;
    /**
     * 断开连接原因
     */
    private DisconnectStatus disconnectStatus;
    /**
     * 连接
     */
    @JsonIgnore
    private Connection connection;
    /**
     * 订阅的topic
     */
    @JsonIgnore
    private Set<SubscribeTopic> topics;
    /**
     * 计数器
     */
    @JsonIgnore
    private transient AtomicInteger atomicInteger;

    public void saveQos2Cache(Integer messageId, PublishMessage publishMessage) {
        qosCache.put(messageId, publishMessage);
    }


    public PublishMessage removeQos2Cache(Integer messageId) {
        return qosCache.remove(messageId);
    }

    public static MqttChannel init(Connection connection) {
        MqttChannel mqttChannel = new MqttChannel();
        mqttChannel.setTopics(new CopyOnWriteArraySet<>());
        mqttChannel.setAtomicInteger(new AtomicInteger(0));
        mqttChannel.setActiveTime(System.currentTimeMillis());
        mqttChannel.setConnection(connection);
        mqttChannel.setAddress(connection.address().toString().substring(1));
        mqttChannel.setId(ContextHolder.getReceiveContext().getIntegrate().getGlobalCounter("channel-id").incrementAndGet());
        connection.channel().pipeline().addFirst(new MyInboundIdleStateHandler(2000, (s) -> {
            if (!connection.isDisposed()) {
                connection.dispose();
            }
        }));
        return mqttChannel;
    }


    public void close() {
        if (!this.connection.isDisposed()) {

            this.connection.dispose();
        }
    }

    public void registryClose(Consumer<MqttChannel> consumer) {
        this.connection.onDispose(() -> consumer.accept(this));
    }


    public int generateMessageId() {
        int index = atomicInteger.incrementAndGet();
        if (index > MAX_MESSAGE_ID) {
            synchronized (this) {
                index = atomicInteger.incrementAndGet();
                if (index > MAX_MESSAGE_ID) {
                    index = 1;
                    atomicInteger.set(index);
                }
            }
        }
        return index;
    }


    @Data
    public static class Auth {

        private String username;

        private byte[] password;


    }

    @Data
    @Builder
    public static class Will {

        private boolean isRetain;

        private String willTopic;

        private MqttQoS mqttQoS;

        private byte[] willMessage;

        public PublishMessage toPublishMessage() {
            PublishMessage publishMessage = new PublishMessage();
            publishMessage.setBody(this.willMessage);
            publishMessage.setTopic(this.willTopic);
            publishMessage.setRetain(this.isRetain);
            publishMessage.setQos(this.mqttQoS.value());
            return publishMessage;
        }

    }

    public void sendPublish(MqttQoS mqttQoS, PublishMessage message) {
        message.setMessageId(this.generateMessageId());
        switch (mqttQoS) {
            case AT_MOST_ONCE:
                this.write(MqttMessageUtils.buildPublish(false, mqttQoS, message.isRetain(), message.getMessageId(),
                        message.getTopic(), PooledByteBufAllocator.DEFAULT.buffer().writeBytes(message.getBody())));
                break;
            case EXACTLY_ONCE:
            case AT_LEAST_ONCE:
            default:
                RetryManager retryManager = ContextHolder.getReceiveContext().getRetryManager();
                RetryMessage retryMessage = new RetryMessage(message.getMessageId(), System.currentTimeMillis(), message.isRetain(),
                        message.getTopic(), MqttQoS.valueOf(message.getQos()), message.getBody(), this);
                retryManager.doRetry(this, retryMessage);
                this.write(MqttMessageUtils.buildPublish(false, mqttQoS, message.isRetain(), message.getMessageId(),
                        message.getTopic(), PooledByteBufAllocator.DEFAULT.buffer().writeBytes(message.getBody())));
                break;
        }
    }


    public void sendRetry(RetryMessage retryMessage) {
        this.write(retryMessage.buildPublishMessage());
    }


    /**
     * write message
     *
     * @param mqttMessage #{@link MqttMessage}
     */
    public void write(MqttMessage mqttMessage) {
        if (this.connection.channel().isActive() && this.connection.channel().isWritable()) {
            connection.channel().writeAndFlush(mqttMessage);
        }
    }

    /**
     * write message
     *
     * @param mqttMessage #{@link MqttMessage}
     */
    public void write(MqttMessage mqttMessage, Consumer<SignalType> other) {
        if (this.connection.channel().isActive() && this.connection.channel().isWritable()) {
            connection.outbound().sendObject(Mono.just(mqttMessage)).then().doFinally(other).subscribe();
        }
    }


}
